package org.pucc.handle

import org.apache.flink.table.api.StatementSet
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.pucc.bean.ImportConfig
import org.pucc.common.DatabaseConstant

/** ---- ----   ---- ----    ---- ---- */
/** ---- ----   Mysql_DDL   ---- ---- */
/** ---- ----   ---- ----   ---- ---- */

object MysqlHandle{
  var n:Int = 0

  def getMysqlDatabaseDDL(importConfig: ImportConfig): String = {
    s"CREATE DATABASE IF NOT EXISTS ${importConfig.sourceDatabase}"
  }

  def getMysqlTableDDL(importConfig: ImportConfig): String ={
    s"""
      |    CREATE TABLE IF NOT EXISTS ${importConfig.sourceDatabase}.${importConfig.sourceTableName} (
      |       ${importConfig.sourceFields}
      |      ) WITH (
      |       'connector' = 'mysql-cdc',
      |       'username' = '${importConfig.sourceName}',
      |       'password' = '${importConfig.sourcePw}',
      |       'database-name' = '${importConfig.sourceDatabase}',
      |       'table-name' = '${importConfig.sourceTableName}',
      |       'hostname' = '${importConfig.sourceUrl}',
      |       'port' = '${importConfig.sourcePort}',
      |       'server-time-zone' = 'Asia/Shanghai',
      |       'jdbc.properties.failOverReadOnly' = 'false',
      |       'jdbc.properties.autoReconnect' = 'true',
      |       'connection.pool.size' = '2'
      |      )
      |""".stripMargin
  }

  /**
   * 根据导入数据的规则来建立flink任务
   * @param tableEnv flink环境
   * @param importConfig 导入规则
   * @return
   */
  def startMysqlJob(tableEnv:StreamTableEnvironment,ss: StatementSet, importConfig: ImportConfig) = {
    /** --- 源数据 --- */
    tableEnv.executeSql(getMysqlDatabaseDDL(importConfig))
    tableEnv.executeSql(getMysqlTableDDL(importConfig))

    /** --- Hudi --- */
    tableEnv.executeSql(HudiHandle.getHudiDatabaseDDL(importConfig))
    tableEnv.executeSql(HudiHandle.getHudiTableDDL(importConfig))

    /** --- 开启实时同步 --- */
    ss.addInsertSql(
      s"""
         |insert into ${importConfig.sinkDatabase}.${importConfig.sinkTableName}
         |select ${importConfig.insertFields}
         |from ${importConfig.sourceDatabase}.${importConfig.sourceTableName}
         |""".stripMargin)

  }

}
